package com.ksyun.campus.metaserver.services.impl;

import com.ksyun.campus.metaserver.client.DataServerClient;
import com.ksyun.campus.metaserver.config.MinfsProperties;
import com.ksyun.campus.metaserver.domain.DataServerInfo;
import com.ksyun.campus.metaserver.services.IHeartbeatService;
import com.ksyun.campus.metaserver.storage.MetadataStore;
import com.ksyun.campus.metaserver.zk.ZooKeeperService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.List;

/**
 * DataServer心跳监控服务实现类
 * 负责监控所有DataServer的健康状态
 * 通过心跳机制检测DataServer是否在线，并处理离线的DataServer
 */
@Slf4j
@Service
public class HeartbeatServiceImpl implements IHeartbeatService {
    
    /**
     * ZooKeeper服务，用于获取和更新DataServer信息
     * 以及判断当前节点是否为主节点
     */
    @Autowired
    private ZooKeeperService zooKeeperService;
    
    /**
     * 元数据存储服务，用于持久化DataServer信息
     */
    @Autowired
    private MetadataStore metadataStore;
    
    /**
     * DataServer客户端，用于与DataServer通信
     * 检查DataServer健康状态
     */
    @Autowired
    private DataServerClient dataServerClient;
    
    /**
     * 系统配置属性，包含心跳相关配置
     */
    @Autowired
    private MinfsProperties properties;
    
    /**
     * 检查DataServer心跳
     * 定时任务，根据配置的时间间隔自动执行
     * 遍历所有已注册的DataServer，检查其健康状态
     * 如果超过配置的超时时间，则标记为不健康
     * 只有在当前节点为主节点时才会执行
     */
    @Scheduled(fixedRateString = "${minfs.heartbeat.check-interval-seconds:10}000")
    @Override
    public void checkHeartbeats() {
        if (!zooKeeperService.isLeader()) {
            log.debug("Not leader, skipping heartbeat check");
            return;
        }
        
        try {
            List<DataServerInfo> dataServers = zooKeeperService.getCurrentDataServers();
            int timeoutSeconds = properties.getHeartbeat().getTimeoutSeconds();
            
            for (DataServerInfo ds : dataServers) {
                checkDataServerHealth(ds, timeoutSeconds);
            }
            
            log.debug("Checked heartbeats for {} data servers", dataServers.size());
        } catch (Exception e) {
            log.error("Error checking heartbeats", e);
        }
    }
    
    /**
     * 检查单个DataServer的健康状态
     * 通过ping和获取健康信息来判断DataServer是否健康
     * 如果健康，则更新容量和使用信息
     * 如果不健康，则标记为不活跃
     *
     * @param ds DataServer信息
     * @param timeoutSeconds 心跳超时时间（秒）
     */
    private void checkDataServerHealth(DataServerInfo ds, int timeoutSeconds) {
        try {
            // 检查DataServer是否响应
            boolean isHealthy = dataServerClient.ping(ds);
            
            if (isHealthy) {
                // 如果需要，更新健康信息
                DataServerClient.DataServerHealthInfo healthInfo = dataServerClient.getHealth(ds);
                if (healthInfo.isHealthy()) {
                    // 更新容量和使用信息
                    ds.setCapacityBytes(healthInfo.getCapacityBytes());
                    ds.setUsedBytes(healthInfo.getUsedBytes());
                    ds.setLastSeen(LocalDateTime.now());
                    ds.setStatus(DataServerInfo.DataServerStatus.ACTIVE);
                    
                    // 更新ZK和本地存储
                    zooKeeperService.updateDataServerInfo(ds);
                    metadataStore.putDataServer(ds);
                    
                    log.debug("DataServer {} is healthy", ds.getDsId());
                } else {
                    markDataServerUnhealthy(ds);
                }
            } else {
                markDataServerUnhealthy(ds);
            }
            
        } catch (Exception e) {
            log.warn("Error checking health for DataServer {}: {}", ds.getDsId(), e.getMessage());
            markDataServerUnhealthy(ds);
        }
    }
    
    /**
     * 标记DataServer为不健康状态
     * 如果DataServer超过配置的超时时间未发送心跳，则标记为不活跃
     *
     * @param ds 要标记的DataServer信息
     */
    private void markDataServerUnhealthy(DataServerInfo ds) {
        try {
            if (!ds.isHealthy(properties.getHeartbeat().getTimeoutSeconds())) {
                log.warn("DataServer {} marked as unhealthy", ds.getDsId());
                ds.setStatus(DataServerInfo.DataServerStatus.INACTIVE);
                metadataStore.putDataServer(ds);
                
                // 注意：ZK临时节点在连接断开时会自动移除
            }
        } catch (Exception e) {
            log.error("Error marking DataServer {} as unhealthy", ds.getDsId(), e);
        }
    }
    
    /**
     * 清理离线的DataServer记录
     * 定时任务，每5分钟执行一次
     * 删除长时间未发送心跳的DataServer记录
     * 并触发相关块的副本修复流程
     * 只有在当前节点为主节点时才会执行
     */
    @Scheduled(fixedRate = 300000) // 每5分钟
    @Override
    public void cleanupOfflineDataServers() {
        if (!zooKeeperService.isLeader()) {
            return;
        }
        
        try {
            List<DataServerInfo> allDataServers = metadataStore.getAllDataServers();
            if (allDataServers.isEmpty()) {
                log.debug("No DataServers found, skipping cleanup");
                return;
            }
            
            LocalDateTime cutoffTime = LocalDateTime.now().minusMinutes(30); // 30分钟离线
            
            for (DataServerInfo ds : allDataServers) {
                if (ds.getLastSeen() != null && ds.getLastSeen().isBefore(cutoffTime)) {
                    if (ds.getStatus() == DataServerInfo.DataServerStatus.INACTIVE) {
                        log.info("Cleaning up long-offline DataServer: {}", ds.getDsId());
                        // 不立即删除 - 让FSCK先处理块复制
                        ds.setStatus(DataServerInfo.DataServerStatus.DECOMMISSIONED);
                        metadataStore.putDataServer(ds);
                    }
                }
            }
        } catch (Exception e) {
            log.error("Error cleaning up offline DataServers", e);
        }
    }
}
